
#include <sstream>
#include "MysqlStorage.h"
#include "define.h"
#include <mqPool.h>
#include <viva.h>
#include "Debug.h"
#include "define.h"
#include "Defines.h"
#include <sstream>
using namespace std;

extern "C" IStorage * create_storage(const char *config) {
	IStorage *obj = new MySQLStorage();
	if(obj->Initialize(config))
		return obj;
	delete obj;
	return NULL; 
}

extern "C" void destroy_storage(MySQLStorage *obj) {
		if(obj != NULL)
			delete obj;
		obj = NULL;
}
///////////////////////////////////////////////////////////

bool MySQLStorage :: Initialize(const char *conf) 
{
 DEBUG2("MqPool::Instance ", conf );
	MqPool *pA =   MqPool::Instance(conf); 	
	if(pA == NULL)
        {
                ERRO <<" MqPool Instance error" << endl;
		return false;
        } 
	else
	{

DEBUG1("start to check MySQL");
		int fc = pA->GetFarmsCounts();
		for(int i=0; i<fc; i++)
		{
			MYSQL* rcon = pA->GetReader(i);
			if(rcon == NULL)
			{
				cerr<<"Get reader by farmid="<< i<<", but return is NULL" << endl;
				pA->CloseReader(i, rcon);
				return false;
			}
                	pA->CloseReader(i, rcon);
		}

		GKeyFile *kv = g_key_file_new();
		GError *err = NULL;
		if(TRUE != g_key_file_load_from_file(kv,conf, G_KEY_FILE_NONE , &err))
		{
			if (err != NULL)
			{
				cerr << "Unable to read file:" << err->message << endl;
				g_error_free (err);
			}
			g_key_file_free(kv);
			return false;
		}

		gchar *pval =  g_key_file_get_value(kv,"global", "writedb",  NULL);
		if(pval == NULL)
		{
			ERRO<<"needn't write to db!"<< endl;
			m_bWrite2DB = false;
		}
		else
		{
			atoi((char *)pval) > 0 ? m_bWrite2DB = true: m_bWrite2DB = false; 
			g_free(pval);
		}
		pval =  g_key_file_get_value(kv,"global", "txlog",  NULL);
		if(pval == NULL)
		{
			cerr<<"no txlog in global"<< endl;
                        m_pTx = NULL;
		}
		else
		{ 
			string txconf = pval;
			g_free(pval);
			m_pTx = new  Viva();
			if(!m_pTx->Initilize(txconf.c_str()))
			{
				delete m_pTx;
				m_pTx = NULL;
				g_key_file_free(kv);
				return false;
			} 
		}  
		g_key_file_free(kv);
		DEBUG1("GREAT!!!!!!!!, Object initialize OK "); 
		return true;
	}
}


int MySQLStorage :: Read(int ihash, GHashTable *kv)
{
 /* read data from db by each key in kv , the result also save in kv.
 if no data, then remove the key from kv.
 in kv, value is not true, it will not be used forever!
*/
        if(kv == NULL)
            return PANDA_SUCCESS;

        GHashTable *h = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, g_free);
	GHashTableIter iter;
	gpointer keyit, valueit;
	g_hash_table_iter_init (&iter, kv);
	while (g_hash_table_iter_next (&iter, &keyit, &valueit))
             g_hash_table_insert(h, g_strdup((char *)keyit), g_strdup((char *)valueit));
        g_hash_table_remove_all(kv);
 
	MqPool *pAn = MqPool::Instance();
	if(pAn == NULL )
	{
		g_hash_table_destroy(h);
		return PANDA_ERR_SYSTEM;
	}
	ihash = ihash%(1+pAn->GetMaxKey());
	int fId = pAn->GetIDbyKey(ihash);
	MYSQL  *conn = pAn->GetReader(fId);
	if(conn == NULL)
	{
		pAn->CloseReader(fId, conn);
		cerr<<"can't get mysql reader connection!"<<endl;
		g_hash_table_destroy(h);
		return PANDA_ERR_SYSTEM;
	}
        int iRet = read_kv_value(conn, ihash, h, kv);
        g_hash_table_destroy(h);
        pAn->CloseReader(fId, conn);
        if(iRet >= 0 )
                return iRet;
        else
                return PANDA_ERR_SYSTEM;
}

int MySQLStorage :: Write(int ihash, GHashTable *kv)
{
 /* all data in kv should be saved into db 
 *  in kv, the key is string, but value is s binary string: unsigned + string.
 *  means string length and really string. 
 */
        long lt = time(NULL);
        int iret = 0;
        if(m_pTx != NULL)
	{
		char sbuf[128];
		int ilen = sprintf(sbuf, "%ld", lt);
		sbuf[ilen] = '\0';
#ifdef  _THREAD_SAFE
		pthread_mutex_lock(&_lock);
#endif
                iret = write_viva(ihash, kv, true, (const char *)sbuf, PANDA_CMD_REPLACE);
#ifdef  _THREAD_SAFE
                pthread_mutex_unlock(&_lock);
#endif
        }
        if(!m_bWrite2DB)
              return iret;

	MqPool *pAn = MqPool::Instance();
	if(pAn == NULL )
		return PANDA_ERR_SYSTEM;
	
	ihash = ihash%(1+pAn->GetMaxKey());
	int fId = pAn->GetIDbyKey(ihash);
	MYSQL  *conn = pAn->GetWriter(fId, 0);
	if(conn == NULL)
	{
		pAn->CloseWriter(fId, conn);
		cerr<<"can't get mysql reader connection!"<<endl;
		return PANDA_ERR_SYSTEM;
	}
	GHashTableIter iter;
	gpointer key, value;
	g_hash_table_iter_init (&iter, kv);
	while (g_hash_table_iter_next (&iter, &key, &value))
		write_kv_value(conn, ihash, (const char *)key, (const char *)value, lt);	
	pAn->CloseWriter(fId, conn);
	return PANDA_SUCCESS;
}


int MySQLStorage :: Atom(int ihash, const char *key, int &ival, bool bDatabase)
{
/* bDatabase means is you want to operate database,
 * if false, then system only write info to viva and return true, ival will not be changed.
 * else (true), system will update/insert database and return the new value for ival. 
 */
        long lt = time(NULL);
        int iret = 0;
        if(m_pTx != NULL)
	{
		char sbuf[128];
		int ilen = sprintf(sbuf, "%ld", lt);
		sbuf[ilen] = '\0';
		char vbuf[128];
		ilen = sprintf(vbuf, "%d", ival);
		vbuf[ilen] = '\0';
#ifdef  _THREAD_SAFE
		pthread_mutex_lock(&_lock);
#endif                
                iret = write_viva(ihash, key,(const char *) vbuf, (const char *)sbuf, PANDA_CMD_ATOM);
#ifdef  _THREAD_SAFE
                pthread_mutex_unlock(&_lock);
#endif
        }
        if(!bDatabase)
              return PANDA_SUCCESS;

        MqPool *pAn = MqPool::Instance();
        if(pAn == NULL )
                return PANDA_ERR_SYSTEM;
        ihash = ihash%(1+pAn->GetMaxKey());
        int fId = pAn->GetIDbyKey(ihash);
        MYSQL  *conn = pAn->GetWriter(fId, 0);
        if(conn == NULL)
        {
                pAn->CloseWriter(fId, conn);
                cerr<<"can't get mysql writer connection!"<<endl;
                return PANDA_ERR_SYSTEM;
        }

	char k[ 2* strlen(key) +1 ];
	int elen = mysql_escape_string(k, key, strlen(key));
	k[elen] = '\0';

	ostringstream sql;

	sql<<"SELECT "<<PANDA_DB_FIELD_VAL<<" FROM " << PANDA_DB_T_HEADER << ihash <<"  WHERE " << PANDA_DB_FIELD_KEY<<"='"<<key <<"'";
	DEBUG2("sql", sql.str().c_str());
	if(mysql_real_query(conn, sql.str().c_str(), sql.str().length() ) != 0){
		cerr << mysql_error(conn) << endl;
		pAn->CloseWriter(fId, conn);
		return PANDA_ERR_SYSTEM;
	}
	MYSQL_RES *result;
	MYSQL_ROW row;
	result = mysql_store_result(conn);
	if(result == NULL){
		cerr << mysql_error(conn) << endl;
		pAn->CloseWriter(fId, conn);
		return PANDA_ERR_SYSTEM;
	}
	long rownum = mysql_num_rows(result);
	int iival = 0; 
	if (rownum > 0 && (row = mysql_fetch_row(result)))
		iival = atoi(row[0]);
	mysql_free_result(result);

	if(!m_bWrite2DB)
	{
		ival += iival;
		pAn->CloseWriter(fId, conn);
		return PANDA_SUCCESS;
	}

	sql.str("");
	if(rownum > 0)
		sql<<"UPDATE "<< PANDA_DB_T_HEADER << ihash <<" SET "<<PANDA_DB_FIELD_VAL<<"="<<ival<<"+"<<PANDA_DB_FIELD_VAL<<","<< PANDA_DB_FIELD_TIME <<"="<< lt <<" WHERE " << PANDA_DB_FIELD_KEY<<"='"<<key <<"'";
	else
		sql<<"INSERT INTO "<< PANDA_DB_T_HEADER<<ihash <<"("<< PANDA_DB_FIELD_KEY<<","<<PANDA_DB_FIELD_VAL<<","<<PANDA_DB_FIELD_TIME<<") VALUES('"<<key <<"','"<< ival <<"',"<<  lt <<")";
	ival += iival;
	DEBUG1(sql.str().c_str());

	if( mysql_real_query(conn, sql.str().c_str(), sql.str().length()) == 0)
	{
		pAn->CloseWriter(fId, conn);
		return PANDA_SUCCESS;
	}
	ERRO <<mysql_errno(conn)<<":"<< mysql_error(conn) << endl;
	pAn->CloseWriter(fId, conn);
	return PANDA_ERR_SYSTEM ;
}


int MySQLStorage :: Delete(int ihash, GHashTable *kv)
{
/* all data in kv should be used to delete from database. 
 * in kv, the key is string, but value is s binary string: unsigned + string.
 * means string length and really string. 
 * in fact, value's string is the save to key string.
 */
	long lt = time(NULL);
	int iret = 0;
	if(m_pTx != NULL)
	{
        char sbuf[128];
        int ilen = sprintf(sbuf, "%ld", lt);
        sbuf[ilen] = '\0';
      
#ifdef  _THREAD_SAFE
                pthread_mutex_lock(&_lock);
#endif
                iret = write_viva(ihash, kv, true, (const char *)sbuf, PANDA_CMD_RM);
#ifdef  _THREAD_SAFE
                pthread_mutex_unlock(&_lock);
#endif
        }
      if(!m_bWrite2DB)
             return iret;

	MqPool *pAn = MqPool::Instance();
	if(pAn == NULL )
		return PANDA_ERR_SYSTEM;
	
	ihash = ihash%(1+pAn->GetMaxKey());
	int fId = pAn->GetIDbyKey(ihash);
	MYSQL  *conn = pAn->GetWriter(fId, 0);
	if(conn == NULL)
	{
		pAn->CloseWriter(fId, conn);
		cerr<<"can't get mysql reader connection!"<<endl;
		return PANDA_ERR_SYSTEM;
	}
        ostringstream sql, subsql;
	sql <<"DELETE FROM " << PANDA_DB_T_HEADER << ihash << " WHERE ";
	GHashTableIter iter;
	gpointer key, value;
	g_hash_table_iter_init (&iter, kv);
        int t = 0; 
	while (g_hash_table_iter_next (&iter, &key, &value))
	{
		if(t > 0)
			subsql<<",";

		char k[ 2*strlen((char *)key) +1 ];
		int elen = mysql_escape_string(k,(char *)key, strlen((char *)key));
		k[elen] = '\0';

		subsql << "'"<< (char *)key <<"'";
		t++;
	}
	if(t == 1)
                sql << PANDA_DB_FIELD_KEY << " ="<< subsql.str(); 
	else if(t > 1)
		sql << PANDA_DB_FIELD_KEY << " IN ("<< subsql.str() << ")";              
	DEBUG1(sql.str().c_str());
	if( mysql_real_query(conn, sql.str().c_str(), sql.str().length()) == 0)
	{
		pAn->CloseWriter(fId, conn);
		return PANDA_SUCCESS;
	}
	ERRO <<mysql_errno(conn)<<":"<< mysql_error(conn) << endl;
	pAn->CloseWriter(fId, conn);
	return PANDA_ERR_SYSTEM ;
}

int MySQLStorage:: SQL(int method, const char *data, unsigned dlen, char **wbuf,unsigned &wsize, unsigned &wbytes)
{
	unsigned uint_len = sizeof(uint32_t);
	int ihash = 0;
	bcopy(data, &ihash, uint_len);
	switch(method)
	{
		case PANDA_SQL_MPL:
			return do_MPL(ihash, data+uint_len, dlen-uint_len, wbuf, wsize, wbytes);
		case PANDA_SQL_PQL:
			return do_PQL(ihash, data+uint_len, dlen-uint_len, wbuf, wsize, wbytes);
		case PANDA_SQL_CPL:
			return do_CPL(ihash, data+uint_len, dlen-uint_len, wbuf, wsize, wbytes);
		case PANDA_SQL_SQL:
			return do_SQL(ihash, data+uint_len, dlen-uint_len, wbuf, wsize, wbytes);
		default:
			return PANDA_ERR_INPUT_DATA;
	}
}

int MySQLStorage :: read_kv_value(MYSQL *conn, int ihash, GHashTable *rkv, GHashTable *kv)
/*
 *  rkv表示要读取的key列表，而kv保存所要读取的数据
 */
{
	MYSQL_RES *result;
	MYSQL_ROW row;

	ostringstream sql, subsql;
	sql <<"SELECT "<< PANDA_DB_FIELD_KEY << ", "<< PANDA_DB_FIELD_VAL <<" FROM " << PANDA_DB_T_HEADER << ihash << " WHERE ";

	GHashTableIter iter;
	gpointer key, value;
	g_hash_table_iter_init (&iter, rkv);
        int t = 0; 
	while (g_hash_table_iter_next (&iter, &key, &value))
	{
		if(t > 0)
			subsql<<",";

		char k[ 2*strlen((char *)key) +1 ];
		int elen = mysql_escape_string(k,(char *)key, strlen((char *)key));
		k[elen] = '\0';

		subsql << "'"<< (char *)key <<"'";
		t++;
	}
	if(t == 1)
                sql << PANDA_DB_FIELD_KEY << " ="<< subsql.str(); 
        else if(t > 1)
                sql << PANDA_DB_FIELD_KEY << " IN ("<< subsql.str() << ")";              
 
        DEBUG2("sql=", sql.str().c_str());
        if(mysql_real_query(conn, sql.str().c_str(), sql.str().length()) != 0){
                ERRO << mysql_error(conn) << endl;
                return -1;
        }

        result = mysql_store_result(conn);
        if(result == NULL){
                ERRO << mysql_error(conn) << endl;
                return  -1;
        }

        long rownum = mysql_num_rows(result);
        if (rownum == 0 )
        {
                mysql_free_result(result);
                return 0;
        }
        DEBUG2("result rows= ", rownum);
        while((row = mysql_fetch_row(result)))
        {
                DEBUG3(row[0]," = ", row[1]);
                g_hash_table_insert(kv, g_strdup(row[0]), g_strdup(row[1]));
        }
        mysql_free_result(result);
        return rownum;
}

int  MySQLStorage :: write_kv_value(MYSQL *conn, int ihash, const char *key, const char *val, long mt)
{
	char k[ 2* strlen(key) +1 ];
	int elen = mysql_escape_string(k,key, strlen(key));
	k[elen] = '\0';
        int vlen = 0;
        bcopy(val, &vlen, sizeof(unsigned));
	char v[ 2* vlen +1 ];
	elen = mysql_escape_string(v, val + sizeof(unsigned) , vlen);
	v[elen] = '\0';

	ostringstream sql;
	sql <<"UPDATE "<< PANDA_DB_T_HEADER << ihash <<" SET "<< PANDA_DB_FIELD_VAL<< " ='"<< v <<"', "<< PANDA_DB_FIELD_TIME <<"="<<mt<<" WHERE "<<PANDA_DB_FIELD_KEY <<"='"<< k <<"'";
 
        DEBUG2("sql=", sql.str().c_str());
        if(mysql_real_query(conn, sql.str().c_str(), sql.str().length()) != 0){
                ERRO << mysql_error(conn) << endl;
                return -1;
        }
        if( mysql_affected_rows(conn) >= 1)
                return 1;
	sql.str("");
	sql<<"INSERT INTO "<< PANDA_DB_T_HEADER<<ihash <<"("<< PANDA_DB_FIELD_KEY<<","<<PANDA_DB_FIELD_VAL<<","<<PANDA_DB_FIELD_TIME<<") VALUES('"<<k <<"', '"<< v <<"',"<<   mt <<")";
 
        if(mysql_real_query(conn, sql.str().c_str(), sql.str().length()) != 0){
                ERRO << mysql_error(conn) << endl;
                return -1;
        }
        return 1;
}


int MySQLStorage :: write_viva(int ihash, GHashTable *kv, bool biStart, const char *stime, const char *opt)
{
/* if biStart is true, means that kv's value is a binary string, it start with unsinged, which is really value's length.
 * if biStart is false, means that kv's value is really string.  
 *
 */
        m_pTx->BeginTX();
        m_pTx->AddData(PANDA_CMD_KEY, opt);
        GHashTableIter iter;
        gpointer key, value;
        g_hash_table_iter_init (&iter, kv);
	while (g_hash_table_iter_next (&iter, &key, &value))
	{
		if(biStart)
		{
			unsigned vlen =  0;
			bcopy(value, &vlen, sizeof(unsigned));
			char *val = g_strndup((char *)value+sizeof(unsigned), vlen);
			m_pTx->AddData((char *)key, val);
			DEBUG4("write data to viva: ", (char *)key, " = ", (char *)val);
			g_free(val);
		}
		else
		{  
			m_pTx->AddData((char *)key, (char *)value);
			DEBUG4("write data to viva: ", (char *)key, " = ", (char *)value);
		} 
	}
        char *shash = g_strdup_printf("%d", ihash);
        m_pTx->AddData(PANDA_DB_HASH_KEY, shash);
	m_pTx->AddData(PANDA_MOD_TIME, stime);
	g_free(shash);    
	if(m_pTx->CommitToLog())
		return PANDA_SUCCESS;
	return PANDA_ERR_SYSTEM;
}

int MySQLStorage :: write_viva(int ihash, const char *key, const char *val, const char *stime, const char *opt)
{
        if(m_pTx == NULL)
                return PANDA_ERR_SYSTEM;
        m_pTx->BeginTX();
        char *shash = g_strdup_printf("%d", ihash);
        m_pTx->AddData(PANDA_DB_HASH_KEY, shash);
	g_free(shash);    
	
        m_pTx->AddData(PANDA_CMD_KEY, opt); // PANDA_CMD_REPLACE);
        m_pTx->AddData(key, val);
	DEBUG4("write data to viva: ", key, " = ", val);
        m_pTx->AddData(PANDA_MOD_TIME, stime);
        if(m_pTx->CommitToLog())
                return PANDA_SUCCESS;
        return PANDA_ERR_SYSTEM;
}

int MySQLStorage:: do_MPL(int ihash, const char *data, unsigned dlen, char **wbuf,unsigned &wsize, unsigned &wbytes)
{
        unsigned uint_len = sizeof(unsigned);
	unsigned ilen = 0;
	int argc = 0;
	bcopy(data, &argc, uint_len);
	ilen += uint_len;
	MYSQL *conn = NULL;
        int fId = 0;
        bool dbclose = false;
        wbytes = 2* uint_len;
 	if(wsize < DATA_BUFFER_SIZE)
	{
		*wbuf = (char *)g_realloc(*wbuf,  DATA_BUFFER_SIZE );
		if(*wbuf == NULL)
		{
			wsize = wbytes = 0;
			return PANDA_ERR_SYSTEM;
		} 
		wsize = DATA_BUFFER_SIZE;
	}

	MqPool *pAn = NULL; 
	if(m_bWrite2DB)
	{
		pAn = MqPool::Instance();
		if(pAn != NULL )
		{
			ihash = ihash%(1+pAn->GetMaxKey());
			fId = pAn->GetIDbyKey(ihash);
			conn = pAn->GetWriter(fId, 0);
                        dbclose = true;
		}
	}
	long lt = 0;
	int iret = PANDA_SUCCESS;
        string s_modtime;
	if(m_pTx != NULL)
	{
		char sbuf[128];
		lt = time(NULL);
		int ilen = sprintf(sbuf, "%ld", lt);
		sbuf[ilen] = '\0';
		s_modtime = sbuf;
	}
	for(int i=0; i< argc; i++)
	{
		int iderror = 0, iverror = 0;
		int slen = 0;
		bcopy(data + ilen, &slen, uint_len);
		ilen += uint_len;
		if(ilen + slen > dlen )
		{
			DEBUG2("ilen+slen>dlen ; slen=", slen);
			iret = PANDA_ERR_INPUT_DATA;
			break;
		}
		char *mpl = (char *)data + ilen;
		if(conn != NULL) 
                     if(execute_pql(conn, mpl, slen) < 0) 
                                iderror  = PANDA_ERR_SYSTEM;
		if(m_pTx != NULL)
		{
			char *s_mpl = g_strndup(mpl, slen);
#ifdef  _THREAD_SAFE
			pthread_mutex_lock(&_lock);
#endif
			iret = write_viva(ihash,PANDA_CMD_MPL, s_mpl, s_modtime.c_str(), PANDA_CMD_MPL);
			if(iret != PANDA_SUCCESS)
				iverror == PANDA_ERR_SYSTEM;
#ifdef  _THREAD_SAFE
			pthread_mutex_unlock(&_lock);
#endif
			g_free(s_mpl);
		}
                if(iderror != 0 || iverror != 0)
                {
                     iderror = PANDA_ERR_SYSTEM;
                     bcopy(&iderror, *wbuf + 2*uint_len + i*uint_len, uint_len);
                }
                else
                      bcopy(&iderror, *wbuf + 2*uint_len + i*uint_len, uint_len);
		ilen += slen;
	}

	if(dbclose)
		pAn->CloseWriter(fId, conn);

	wbytes = (argc+2)*uint_len ;
	bcopy(&wbytes,  *wbuf, uint_len);
	bcopy(&argc,  *wbuf + uint_len, uint_len);
	return PANDA_SUCCESS;
}

int MySQLStorage :: do_SQL(int ihash, const char *data, unsigned dlen, char **wbuf,unsigned &wsize, unsigned &wbytes)
{
	if(wsize < DATA_BUFFER_SIZE)
	{
		*wbuf = (char *)g_realloc(*wbuf,  DATA_BUFFER_SIZE );
		if(*wbuf == NULL)
		{
			wsize = wbytes = 0;
			return PANDA_ERR_SYSTEM;
		} 
		wsize = DATA_BUFFER_SIZE;
	}
	unsigned uint_len = sizeof(unsigned);
	unsigned ilen = 0;
	MYSQL *conn = NULL;
        int fId = 0;
        bool dbclose = false;
        wbytes = 2* uint_len;
	MqPool *pAn = NULL; 
	
	pAn = MqPool::Instance();
	if(pAn != NULL )
	{
		ihash = ihash%(1+pAn->GetMaxKey());
		fId = pAn->GetIDbyKey(ihash);
		conn = pAn->GetReader(fId);
		dbclose = true;
	}
	int iret = PANDA_SUCCESS;
        int slen = 0;
	bcopy(data + ilen, &slen, uint_len);
	ilen += uint_len;
	if(ilen + slen > dlen )
	{
		DEBUG2("ilen+slen>dlen ; slen=", slen);
		iret = PANDA_ERR_INPUT_DATA;
	}
	else
	{
		char *mpl = (char *)data + ilen;
		if(conn != NULL ) 
			iret = fetch_data(conn, mpl, slen, wbuf, wsize, wbytes, true); 
	}

	if(dbclose)
		pAn->CloseReader(fId, conn);
DEBUG6("iret = ", iret, "; wsize =", wsize, "; wbytes=", wbytes);
	if(wbytes < 4*uint_len)
	{ // means no data be get, so set error = NO_DATA
		wbytes  =  2*uint_len ;
		bcopy(&wbytes,  *wbuf, uint_len);
                ilen = PANDA_ERR_NO_DATA;
		bcopy(&ilen,  *wbuf + uint_len, uint_len);
	}
	return PANDA_SUCCESS;
}


int MySQLStorage :: do_PQL(int ihash, const char *data, unsigned dlen, char **wbuf,unsigned &wsize, unsigned &wbytes)
{
        unsigned uint_len = sizeof(unsigned);
	unsigned ilen = 0;
	int argc = 0;
	bcopy(data, &argc, uint_len);
	ilen += uint_len;
	MYSQL *conn = NULL;
        int fId = 0;
        bool dbclose = false;
        wbytes = 2* uint_len;
 	if(wsize < DATA_BUFFER_SIZE)
	{
		*wbuf = (char *)g_realloc(*wbuf,  DATA_BUFFER_SIZE );
		if(*wbuf == NULL)
		{
			wsize = wbytes = 0;
			return PANDA_ERR_SYSTEM;
		} 
		wsize = DATA_BUFFER_SIZE;
	}
	
	MqPool *pAn = NULL; 
	if(m_bWrite2DB)
	{
		pAn = MqPool::Instance();
		if(pAn != NULL )
		{
			ihash = ihash%(1+pAn->GetMaxKey());
			fId = pAn->GetIDbyKey(ihash);
			conn = pAn->GetWriter(fId, 0);
                        dbclose = true;
		}
	}
	long lt = time(NULL);
	int iret = PANDA_SUCCESS;
        string s_modtime;
	if(m_pTx != NULL)
	{
		char sbuf[128];
		int ilen = sprintf(sbuf, "%ld", lt);
		sbuf[ilen] = '\0';
		s_modtime = sbuf;
	}
	for(int i=0; i< argc -1; i++)
	{
		int slen = 0;
		bcopy(data + ilen, &slen, uint_len);
		ilen += uint_len;
		if(ilen + slen > dlen )
		{
			DEBUG2("ilen+slen>dlen ; slen=", slen);
			iret = PANDA_ERR_INPUT_DATA;
			break;
		}
		char *mpl = (char *)data + ilen;
		if(conn != NULL ) 
			execute_pql(conn, mpl, slen);
		if(m_pTx != NULL )
		{
			char *s_mpl = g_strndup(mpl, slen);
#ifdef  _THREAD_SAFE
			pthread_mutex_lock(&_lock);
#endif
			iret = write_viva(ihash,PANDA_CMD_MPL, s_mpl, s_modtime.c_str(), PANDA_CMD_MPL);
#ifdef  _THREAD_SAFE
			pthread_mutex_unlock(&_lock);
#endif
			g_free(s_mpl);
		}
		ilen += slen;
	}

        DEBUG4("wsize = ", wsize, "; wbytes = ", wbytes);  
        int slen = 0;
	bcopy(data + ilen, &slen, uint_len);
	ilen += uint_len;
	if(ilen + slen > dlen )
	{
		DEBUG2("ilen+slen>dlen ; slen=", slen);
		iret = PANDA_ERR_INPUT_DATA;
	}
	else
	{
		char *mpl = (char *)data + ilen;
		if(!m_bWrite2DB)
		{
DEBUG1("run select ,so get db conection!");
			pAn = MqPool::Instance();
			if(pAn != NULL )
			{
				ihash = ihash%(1+pAn->GetMaxKey());
				fId = pAn->GetIDbyKey(ihash);
				conn = pAn->GetWriter(fId, 0);
				dbclose = true;
			}
		}
		if(conn != NULL ) 
			iret = fetch_data(conn, mpl, slen, wbuf, wsize, wbytes, true); 
	}

	if(dbclose)
		pAn->CloseWriter(fId, conn);
DEBUG6("iret = ", iret, "; wsize =", wsize, "; wbytes=", wbytes);
	if(wbytes < 4*uint_len)
	{ // means no data be get, so set error = NO_DATA
		wbytes  =  2*uint_len ;
		bcopy(&wbytes,  *wbuf, uint_len);
                ilen = PANDA_ERR_NO_DATA;
		bcopy(&ilen,  *wbuf + uint_len, uint_len);
	}
	return PANDA_SUCCESS;
}

int MySQLStorage :: do_CPL(int ihash, const char *data, unsigned dlen, char **wbuf,unsigned &wsize, unsigned &wbytes)
{
	if(wsize < DATA_BUFFER_SIZE)
	{ //mybe last realloc fail, so wsize = 0!
		*wbuf = (char *)g_realloc(*wbuf,  DATA_BUFFER_SIZE );
		if(*wbuf == NULL)
		{
			wsize = wbytes = 0;
			return PANDA_ERR_SYSTEM;
		} 
		wsize = DATA_BUFFER_SIZE;
	}
	unsigned uint_len = sizeof(unsigned);
	unsigned ilen = 0;
	int fId = 0;
	MYSQL *conn = NULL;
	bool dbclose = false;
	MqPool *pAn = NULL; 
	if(m_bWrite2DB)
	{
		pAn = MqPool::Instance();
		if(pAn != NULL )
		{
			ihash = ihash%(1+pAn->GetMaxKey());
			fId = pAn->GetIDbyKey(ihash);
			conn = pAn->GetWriter(fId, 0);
                        dbclose = true; //must use Close to release mutex!
		}
	}
	long lt = 0;
	int iret = PANDA_SUCCESS;
        string s_modtime;
	if(m_pTx != NULL)
	{
		char sbuf[128];
		lt = time(NULL);
		int ilen = sprintf(sbuf, "%ld", lt);
		sbuf[ilen] = '\0';
		s_modtime = sbuf;
	}
        int selen=0, inlen=0, uplen = 0;
        char *seql = NULL;
        char *inql = NULL;
        char *upql = NULL;
        ilen = 0;	
	bcopy(data + ilen, &selen, uint_len);
	ilen += uint_len;
	if(ilen + selen > dlen )
	{
		DEBUG2("ilen+selen>dlen ; selen=", selen);
		if(dbclose)
			pAn->CloseWriter(fId, conn);
		return PANDA_ERR_INPUT_DATA;
	}
	seql = (char *)data + ilen;
	ilen += selen;

	bcopy(data + ilen, &uplen, uint_len);
	ilen += uint_len;
	if(ilen + uplen > dlen )
	{
		DEBUG2("ilen+uplen>dlen ; uplen=", uplen);
		if(dbclose)
			pAn->CloseWriter(fId, conn);
		return PANDA_ERR_INPUT_DATA;
	}
	upql = (char *)data + ilen;
	ilen += uplen;

	bcopy(data + ilen, &inlen, uint_len);
	ilen += uint_len;
	if(ilen + inlen > dlen )
	{
		DEBUG2("ilen+inlen>dlen ; inlen=", inlen);
		if(dbclose)
			pAn->CloseWriter(fId, conn);
		return PANDA_ERR_INPUT_DATA;
	}
	inql = (char *)data + ilen;

	if(m_pTx != NULL )
	{
		GHashTable *h = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, g_free);
		g_hash_table_insert(h, g_strdup_printf("%d",1), g_strndup(seql, selen));
		g_hash_table_insert(h, g_strdup_printf("%d",2), g_strndup(upql, uplen));
		g_hash_table_insert(h, g_strdup_printf("%d",3), g_strndup(inql, inlen));
#ifdef  _THREAD_SAFE
		pthread_mutex_lock(&_lock);
#endif
		iret = write_viva(ihash, h, false, s_modtime.c_str(), PANDA_CMD_CPL);
#ifdef  _THREAD_SAFE
		pthread_mutex_unlock(&_lock);
#endif
		g_hash_table_destroy(h);
	}

	if(conn != NULL )
	{ 
		if(fetch_data(conn, seql, selen, wbuf, wsize, wbytes, false) > 0)
                    execute_pql(conn, upql, uplen);
                else
                    execute_pql(conn, inql, inlen);
	}
	if(dbclose)
		pAn->CloseWriter(fId, conn);

	wbytes = 2*uint_len ;
	bcopy(&wbytes,  *wbuf, uint_len);
	bcopy(&iret,  *wbuf + uint_len, uint_len);
	return PANDA_SUCCESS;
}

int MySQLStorage :: execute_pql(MYSQL *conn, const char *sql, unsigned len)
{ // execute PQL: update/delete/insert

	if(mysql_real_query(conn, sql, len ) != 0){
		ERRO << mysql_error(conn) << endl;
		return  PANDA_ERR_SYSTEM;
	}
	int rownum =   mysql_affected_rows(conn);
        return rownum;
}

int MySQLStorage :: fetch_data(MYSQL *conn, const char *sql, unsigned len, char **wbuf, unsigned &wsize, unsigned &wbytes, bool bret)
{ 
/*  length + retcode + row + fields + klen+ key + vlen+ val + ... ...
 *  execute SELECT *
 */ 

	if(mysql_real_query(conn, sql, len ) != 0){
		ERRO << mysql_error(conn) << endl;
                wbytes = 0;
		return  PANDA_ERR_SYSTEM;
	}
/*
 // has been checked at last function!
 
        if(wsize < DATA_BUFFER_SIZE)
        {
            *wbuf = (char *)g_realloc(*wbuf,  DATA_BUFFER_SIZE );
            if(*wbuf == NULL)
            {
                wsize = wbytes = 0;
                return PANDA_ERR_SYSTEM;
            } 
            wsize = DATA_BUFFER_SIZE;
	}

*/
	int rownum = 0;
        int num_fields = 0;
        unsigned uint_len = sizeof(unsigned);
        unsigned used = 4*uint_len;

	MYSQL_RES *result = mysql_store_result(conn);
	if(result != NULL){
		rownum = mysql_num_rows(result);
		if (rownum <= 0 )
		{
			mysql_free_result(result);
			wbytes = 4*uint_len;
			bcopy(&wbytes,  *wbuf, uint_len);
                        int retcode = PANDA_SUCCESS;  
			bcopy(&retcode,  *wbuf + uint_len, uint_len); // retcode
			bcopy(&rownum, *wbuf+2*uint_len, uint_len);
			bcopy(&num_fields , *wbuf+3*uint_len, uint_len);
			return rownum;
		}
		if(bret){
			num_fields = (int) mysql_num_fields(result);
			MYSQL_FIELD *fields = mysql_fetch_fields(result);
			MYSQL_ROW row;
                        int no = 0;
			while((row = mysql_fetch_row(result)))
			{
                                int t = 0;
                                if(no == 0)
				for (t = 0; t < num_fields; t++)
				{
					int ik = strlen((char *)fields[t].name);
					bcopy(&ik, *wbuf+used, uint_len);
					used += uint_len;
					bcopy( fields[t].name, *wbuf+used, ik);
					used += ik;
				}
				for (t = 0; t < num_fields; t++)
				{
					int iv = strlen((char *)row[t]);
					if(wsize < used + uint_len + iv)
					{
						*wbuf = (char *)g_realloc(*wbuf, wsize*2);
						if(*wbuf == NULL)
						{
							wsize = wbytes=0;
							break;
						}
						wsize *=2;         
					}
					bcopy(&iv, *wbuf+used, uint_len);
					used += uint_len;
					bcopy( row[t], *wbuf+used, iv);
					used += iv;
				}
				no++;
			}
		}
		mysql_free_result(result);
	}
	if(wsize > uint_len*4)
	{
		wbytes = used;
		bcopy(&used, *wbuf, uint_len);
		used = 0; // retcode = 0 = PANDA_SUCCESS
		bcopy(&used, *wbuf+uint_len, uint_len);
		bcopy(&rownum, *wbuf+2*uint_len, uint_len);
		bcopy(&num_fields , *wbuf+3*uint_len, uint_len);
	}
	return rownum;
}
